package com.mjf.day3

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.api.common.functions.RichFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

/**
 * 自定义SinkFunction写入到Jdbc
 */
object MyJdbcSink {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream: DataStream[String] = env.readTextFile("D:\\coding\\idea\\flink-stu\\src\\main\\input\\sensor.txt")

    val dataStream: DataStream[SensorReading] = inputStream.map(
      data => {
        val dataArr: Array[String] = data.split(",")
        SensorReading(dataArr(0), dataArr(1).toLong, dataArr(2).toDouble)
      }
    )

    dataStream.addSink(new MyJdbcFunction())

    env.execute("MyJdbcSink")

  }
}

// 自定义一个SinkFunction
class MyJdbcFunction() extends RichSinkFunction[SensorReading]{
  // 首先定义sql连接，以及预编译语句
  var conn: Connection = _
  var insertStmt: PreparedStatement = _
  var updateStmt: PreparedStatement = _

  // 在open生命周期方法中创建连接以及预编译语句
  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://hadoop103:3306/test", "root", "123456")
    insertStmt = conn.prepareStatement("insert into temp (sensor, temperature) values(?,?)")
    updateStmt = conn.prepareStatement("update temp set temperature = ? where sensor = ?")
  }


  override def invoke(value: SensorReading, context: SinkFunction.Context[_]): Unit = {
    // 执行更新语句
    updateStmt.setDouble(1, value.temperature)
    updateStmt.setString(2, value.id)
    updateStmt.execute()
    // 如果刚才没有更新数据，那么执行插入操作
    if (updateStmt.getUpdateCount == 0) {
      insertStmt.setString(1, value.id)
      insertStmt.setDouble(2, value.temperature)
      insertStmt.execute()
    }
  }

  // 关闭操作
  override def close(): Unit = {
    insertStmt.close()
    updateStmt.close()
    conn.close()
  }
}